package com.navinfo.opentsp.platform.computing.analysis.application;

import com.navinfo.opentsp.platform.computing.analysis.entity.CountEntity;
import com.navinfo.opentsp.platform.computing.analysis.entity.ParaBean;
import com.navinfo.opentsp.platform.computing.analysis.entity.PrePoint;
import com.navinfo.opentsp.platform.computing.analysis.entity.VehicleStatusAdditionDto;
import com.navinfo.opentsp.platform.computing.analysis.util.PropertiesUtil;
import com.navinfo.opentsp.platform.computing.analysis.util.RunningTypeEnum;
import com.navinfo.opentsp.platform.computing.analysis.util.TidPartitioner;
import com.navinfo.opentsp.platform.location.protocol.common.LCStatusType;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.*;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.joda.time.DateTime;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import scala.Tuple2;
import scala.Tuple3;
import scala.collection.mutable.WrappedArray;

import java.math.BigDecimal;
import java.text.ParseException;
import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.stream.Collectors;
import java.util.stream.StreamSupport;

/**
 * @Author: wujiangbo
 * @Create: 2018/08/23 下午2:33
 */
public class SummaryDaily {

    private static Logger logger = LoggerFactory.getLogger(SummaryDaily.class);

    public static void main(String[] args) throws ParseException {

        SparkSession.Builder builder = SparkSession.builder();
        final Map<String, String> properties = PropertiesUtil.getProperties();

        if (Boolean.parseBoolean(properties.get("run.env.local"))) {
            builder.master("local");
        }


        final Integer type = Integer.valueOf(properties.get("system.running.type"));
        RunningTypeEnum runningTypeEnum = RunningTypeEnum.valueOf(type);
        logger.error("系统运行类型：{},{}", runningTypeEnum.getType(), runningTypeEnum.getDesc());


        SparkSession spark = builder.appName("SummaryDaily")
                .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
                .enableHiveSupport()
                .getOrCreate();

        properties.forEach((key, value) -> {
            final String hadoop = StringUtils.substringAfter(key, "spark.hadoop.");
            if (StringUtils.isNotEmpty(hadoop)) {
                JavaSparkContext jsc = new JavaSparkContext(spark.sparkContext());
                jsc.hadoopConfiguration().set(hadoop, value);
            }
        });

        // 日期
        String dateStr = args[0];
        int partitionNum = 1 << 10;
        try {
            partitionNum = Integer.parseInt(args[1]);
        } catch (Exception e) {

        }
        if (StringUtils.isNotEmpty(dateStr) && StringUtils.contains(dateStr, "-")) {
            dateStr = dateStr.replace("-", "");
        }

        DateTimeFormatter df = DateTimeFormatter.ofPattern("yyyyMMdd");
        LocalDate startDate = LocalDate.parse(dateStr, df);
        logger.error("date {}", dateStr);
        String day = startDate.format(df);
        logger.error("spark execute trip daily ,execute date {}", day);

        //查询0200原始点
        String sparkSql = null;
        if (type == RunningTypeEnum.DF.getType()) {
            sparkSql = String.valueOf(PropertiesUtil.getProperties("hive.0200.lacation.dongfeng"));
        } else if (type == RunningTypeEnum.QQ.getType()) {
            sparkSql = String.valueOf(PropertiesUtil.getProperties("hive.0200.lacation.qingqi"));
        } else if (type == RunningTypeEnum.YQ.getType()) {
            sparkSql = String.valueOf(PropertiesUtil.getProperties("hive.0200.lacation.yiqi"));
        }else if (type == RunningTypeEnum.HS.getType()) {
            sparkSql = String.valueOf(PropertiesUtil.getProperties("hive.0200.lacation.huashen"));
        }
        sparkSql = String.format(sparkSql, day);
        Dataset<Row> rrr = spark.sql(sparkSql);
        JavaRDD<Row> lines = rrr.toJavaRDD();

        // 查询hive中行程汇总的后的日统计
        String vehicleBasicDataSql = null;
        if (type == RunningTypeEnum.DF.getType() || type == RunningTypeEnum.QQ.getType()) {
            vehicleBasicDataSql = String.valueOf(PropertiesUtil.getProperties("hive.vehicle.basic.data.hql.qingqi"));
        } else if (type == RunningTypeEnum.YQ.getType()) {
            vehicleBasicDataSql = String.valueOf(PropertiesUtil.getProperties("hive.vehicle.basic.data.hql.yiqi"));
        }else if (type == RunningTypeEnum.HS.getType()) {
            vehicleBasicDataSql = String.valueOf(PropertiesUtil.getProperties("hive.vehicle.basic.data.hql.huashen"));
        }
        vehicleBasicDataSql = String.format(vehicleBasicDataSql, day);
        Dataset<CountEntity> vehicleBasic = spark.sql(vehicleBasicDataSql).mapPartitions(
                rowToCountEntity(), Encoders.bean(CountEntity.class));

        JavaPairRDD<Long, CountEntity> vehicleBasicDataPairRDD = vehicleBasic.toJavaRDD()
                .mapPartitionsToPair((PairFlatMapFunction<Iterator<CountEntity>, Long, CountEntity>) iterator -> {
                    Iterable<CountEntity> iterable = () -> iterator;
                    return StreamSupport.stream(iterable.spliterator(), false)
                            .map(s -> new Tuple2<>(s.getTerminalId(), s)).iterator();
                });


        // 创建CountEntity
        Function<ParaBean, CountEntity> createCount = paraBean -> {
            CountEntity countEntity = new CountEntity();
            DateTime dateTime = new DateTime(paraBean.getGpsDate());
            String dataDate = dateTime.toString("yyyy-MM-dd");
            countEntity.setDate(dataDate);
            countEntity.setTerminalId(paraBean.getTid());
            countEntity.setTimestamp(dateTime.withTimeAtStartOfDay().getMillis());
            countEntity.setDate(dataDate);
            countEntity.setTerminalId(paraBean.getTid());
            return countEntity;
        };
        Map<String, Object> kafkaConfig = new HashMap<>();
        kafkaConfig.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, PropertiesUtil.getProperties("kafka.address"));
        kafkaConfig.put(ProducerConfig.RETRIES_CONFIG, PropertiesUtil.getProperties("kafka.retries"));
        kafkaConfig.put(ProducerConfig.BATCH_SIZE_CONFIG, PropertiesUtil.getProperties("kafka.batch.size"));
        kafkaConfig.put(ProducerConfig.LINGER_MS_CONFIG, PropertiesUtil.getProperties("kafka.linger.ms"));
        kafkaConfig.put(ProducerConfig.BUFFER_MEMORY_CONFIG, PropertiesUtil.getProperties("kafka.buffer.memory"));
        kafkaConfig.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                PropertiesUtil.getProperties("kafka.key.serializer"));
        kafkaConfig.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                PropertiesUtil.getProperties("kafka.value.serializer"));
        kafkaConfig.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG,
                PropertiesUtil.getProperties("kafka.request.timeout.ms"));
        String topic = PropertiesUtil.getProperties("topic");
        // 初始化kafka配置
        Function<Map<String, Object>, KafkaTemplate<String, Object>> kafkaTemplateMapFunction =
                v1 -> new KafkaTemplate<>(new DefaultKafkaProducerFactory<String, Object>(v1));

        // init data
        Function<Tuple2<ParaBean, CountEntity>,
                Tuple2<CountEntity, PrePoint>> initData = paraBeanCountEntityTuple2 -> {
            ParaBean paraBean = paraBeanCountEntityTuple2._1;
            CountEntity countEntity = paraBeanCountEntityTuple2._2;
            PrePoint prePoint = new PrePoint();
            prePoint.setGpsDate(paraBean.getGpsDate());
            countEntity.setStart(Long.valueOf(paraBean.getGpsDate() / 1000).intValue());
            prePoint.setLat(paraBean.getLat());
            prePoint.setLng(paraBean.getLng());
            prePoint.setOilValue(paraBean.getOilValue());
            return new Tuple2<>(countEntity, prePoint);
        };

        // 计算 参数，计算结果，前一个点
        VoidFunction<Tuple3<ParaBean, CountEntity, PrePoint>> countData = paraBeanCountEntityPrePointTuple3 -> {
            ParaBean parameter = paraBeanCountEntityPrePointTuple3._1();
            CountEntity countEntity = paraBeanCountEntityPrePointTuple3._2();
            PrePoint cachePoint = paraBeanCountEntityPrePointTuple3._3();
            if (parameter.getSpeed() > 0) {
                countEntity.setSpeedSum(countEntity.getSpeedSum() + parameter.getSpeed());
                countEntity.setSpeedNum(countEntity.getSpeedNum() + 1);
            }

            cachePoint.setGpsDate(parameter.getGpsDate());
            //new add config
            countEntity.setEnd(Long.valueOf(parameter.getGpsDate() / 1000).intValue());
            countEntity.setbLat(cachePoint.getLat());
            countEntity.setbLng(cachePoint.getLng());
            countEntity.seteLat(parameter.getLat());
            countEntity.seteLng(parameter.getLng());

            double oilValue = parameter.getOilValue();
            double cacheOilvalue = cachePoint.getOilValue();
            BigDecimal bd1 = new BigDecimal(Double.toString(cacheOilvalue));
            BigDecimal bd2 = new BigDecimal(Double.toString(oilValue));
            if (oilValue != 0 && bd1.subtract(bd2).doubleValue() > 0) {
                countEntity.setOilValue(bd1.subtract(bd2)
                        .add(new BigDecimal(Double.toString(countEntity.getOilValue())))
                        .doubleValue());
            }
            if (oilValue != 0) {
                cachePoint.setOilValue(oilValue);
            }
        };

        // row -> ParaBean
        FlatMapFunction<Iterator<Row>, ParaBean> iteratorTripDailyFlatMapFunction =
                getIteratorParaBeanFlatMapFunction(type);

        PairFlatMapFunction<Iterator<ParaBean>, Long, List<ParaBean>> pairPartitionFunction1 = tripInfoPoint -> {
            List<ParaBean> paraBeanList = new ArrayList<>();
            tripInfoPoint.forEachRemaining(paraBeanList::add);
            return paraBeanList.stream().
                    collect(Collectors.groupingBy(ParaBean::getTid)).
                    entrySet().stream().
                    map(longListEntry ->
                            new Tuple2<>(longListEntry.getKey(),
                                    longListEntry.getValue()))
                    .collect(Collectors.toList()).iterator();
        };

        FlatMapFunction<Iterator<Tuple2<Long, List<ParaBean>>>,
                CountEntity> iteratorFlatMapFunction = tuple2Iterator -> {
            List<CountEntity> list = new ArrayList<>();
            while (tuple2Iterator.hasNext()) {
                // 每个key初始化数据
                List<ParaBean> paraBeans = new ArrayList<>(tuple2Iterator.next()._2);
                // 升序排序
                paraBeans = paraBeans.stream()
                        .sorted((o1, o2) -> (int) (o1.getGpsDate() - o2.getGpsDate()))
                        .collect(Collectors.toList());

                CountEntity countEntity = null;
                PrePoint prePoint = null;

                for (ParaBean paraBean : paraBeans) {
                    if (countEntity == null) {
                        // 初始化CountEntity
                        countEntity = createCount.call(paraBean);
                        Tuple2<CountEntity, PrePoint> initDataLocal = initData.call(
                                new Tuple2<>(paraBean, countEntity));
                        prePoint = initDataLocal._2;
                    }
                    countData.call(new Tuple3<>(paraBean, countEntity, prePoint));
                }
                if (countEntity != null) {
                    countEntity.setsDate(countEntity.getEnd() - countEntity.getStart());
                    list.add(countEntity);
                }
            }
            return list.iterator();
        };

        // pair rdd (tid,List<TripInfoPoint>)
        lines.mapPartitions(iteratorTripDailyFlatMapFunction)
                .mapPartitionsToPair(pairPartitionFunction1)
                .partitionBy(new TidPartitioner(partitionNum))
                .reduceByKey((paraBeans, paraBeans2) -> {
                    paraBeans2.addAll(paraBeans);
                    return paraBeans2;
                })
                .mapPartitions(iteratorFlatMapFunction)
                .mapPartitionsToPair((PairFlatMapFunction<Iterator<CountEntity>, Long, CountEntity>) iterator -> {
                    Iterable<CountEntity> iterable = () -> iterator;
                    return StreamSupport.stream(iterable.spliterator(), false)
                            .map(s -> new Tuple2<>(s.getTerminalId(), s)).iterator();
                }).join(vehicleBasicDataPairRDD, new TidPartitioner(partitionNum))
                .mapPartitions(pariToCountEntityRDD())
                .foreachPartition(countEntityIterator -> {
                    KafkaTemplate<String, Object> kafkaTemplate = kafkaTemplateMapFunction.call(kafkaConfig);
                    countEntityIterator.forEachRemaining(countEntity -> {
                        kafkaTemplate.send(topic, String.valueOf(countEntity.getTerminalId()), countEntity);
                    });
                });
    }

    private static FlatMapFunction<Iterator<Row>, ParaBean> getIteratorParaBeanFlatMapFunction(int type) {
        FlatMapFunction<Iterator<Row>, ParaBean> iteratorTripDailyFlatMapFunction = null;
        if (type == RunningTypeEnum.DF.getType() || type == RunningTypeEnum.QQ.getType()) {
            iteratorTripDailyFlatMapFunction = rowIterator -> {
                List<ParaBean> list = new ArrayList<>();
                while (rowIterator.hasNext()) {
                    Row row = rowIterator.next();
                    ParaBean paraBean = new ParaBean();

                    //取标准速度
                    BigDecimal speedBigDecimal = new BigDecimal(Long.parseLong(row.getAs("electricvehicle"))).divide(
                            new BigDecimal("100"), 0, BigDecimal.ROUND_HALF_UP);
                    int speed = speedBigDecimal.intValue();
                    if (speed == 0) {
                        speed = Integer.parseInt(row.getAs("speed"));
                    }
                    long gpsDate = Long.parseLong(row.getAs("gpsdate")) * 1000;
                    paraBean.setGpsSpeed(Integer.parseInt(row.getAs("speed")));
                    paraBean.setSpeed(speed);
                    paraBean.setGpsDate(gpsDate);
                    paraBean.setLat(Integer.parseInt(row.getAs("latitude")));
                    paraBean.setLng(Integer.parseInt(row.getAs("longitude")));
                    List<VehicleStatusAdditionDto> statusAdditionDtos = VehicleStatusAdditionDto.parseToList(
                            row.getAs("statusaddition"));
                    for (VehicleStatusAdditionDto dto : statusAdditionDtos) {
                        if (Integer.parseInt(dto.getTypes()) == LCStatusType.StatusType.oilValue.getNumber()) {
                            // 剩余油量
                            double oilValue = dto.getStatusValue() / 100d;
                            paraBean.setOilValue(oilValue);
                        }
                    }
                    paraBean.setTid(Long.parseLong(row.getAs("tid")));
                    list.add(paraBean);
                }
                return list.iterator();
            };
        } else if (type == RunningTypeEnum.YQ.getType()) {
            iteratorTripDailyFlatMapFunction = rowIterator -> {
                List<ParaBean> list = new ArrayList<>();
                while (rowIterator.hasNext()) {
                    Row row = rowIterator.next();
                    ParaBean paraBean = new ParaBean();
                    //取标准速度
                    int speed = row.getAs("speed");
                    long gpsDate = (long) row.getAs("gpsdate") * 1000L;
                    paraBean.setSpeed(speed);
                    paraBean.setGpsDate(gpsDate);
                    paraBean.setLat(row.getAs("latitude"));
                    paraBean.setLng(row.getAs("longitude"));
                    if (row.getAs("statusaddition") != null) {
                        WrappedArray.ofRef<Row> statusaddition = row.getAs("statusaddition");
                        for (Row status : statusaddition.array()) {
                            if (status.getAs("types").equals(LCStatusType.StatusType.oilValue.name())) {
                                // 剩余油量
                                double oilValue = (long) status.getAs("statusvalue") / 100d;
                                paraBean.setOilValue(oilValue);
                            }
                        }
                    }
                    paraBean.setTid(row.getAs("terminalid"));
                    list.add(paraBean);
                }
                return list.iterator();
            };
        }
        return iteratorTripDailyFlatMapFunction;
    }

    private static MapPartitionsFunction<Row, CountEntity> rowToCountEntity() {
        return (MapPartitionsFunction<Row, CountEntity>) iterator -> {
            Iterable<Row> iterable = () -> iterator;
            return StreamSupport.stream(iterable.spliterator(), false)
                    .map(s -> {
                        CountEntity entity = new CountEntity();
                        long terminalId = Long.parseLong(
                                String.valueOf(Optional.ofNullable(s.getAs("terminalId")).orElse(0L)));
                        double mMilage = Double.parseDouble(
                                String.valueOf(Optional.ofNullable(s.getAs("mMilage")).orElse(0D)));
                        double fuel = Double.parseDouble(
                                String.valueOf(Optional.ofNullable(s.getAs("fuel")).orElse(0D)));
                        double workHours = Double.parseDouble(
                                String.valueOf(Optional.ofNullable(s.getAs("workHours")).orElse(0D)));
                        double idleHours = Double.parseDouble(
                                String.valueOf(Optional.ofNullable(s.getAs("idleHours")).orElse(0D)));
                        int stboTimes = Integer.parseInt(
                                String.valueOf(Optional.ofNullable(s.getAs("stboTimes")).orElse(0)));
                        int acceTimes = Integer.parseInt(
                                String.valueOf(Optional.ofNullable(s.getAs("acceTimes")).orElse(0)));
                        int overspeedTimes = Integer.parseInt(
                                String.valueOf(Optional.ofNullable(s.getAs("overspeedTimes")).orElse(0)));
                        int hsnsTimes = Integer.parseInt(
                                String.valueOf(Optional.ofNullable(s.getAs("hsnsTimes")).orElse(0)));
                        int tdTimes = Integer.parseInt(
                                String.valueOf(Optional.ofNullable(s.getAs("tdTimes")).orElse(0)));
                        double fuelRateDay = Double.parseDouble(
                                String.valueOf(Optional.ofNullable(s.getAs("fuelRateDay")).orElse(0D)));
                        double avgSpeedNoidle = Double.parseDouble(
                                String.valueOf(Optional.ofNullable(s.getAs("avgSpeedNoidle")).orElse(0D)));
                        double avgSpeed = Double.parseDouble(
                                String.valueOf(Optional.ofNullable(s.getAs("avgSpeed")).orElse(0D)));
                        double idleP = Double.parseDouble(
                                String.valueOf(Optional.ofNullable(s.getAs("idleP")).orElse(0D)));
                        double avgHsnsDay = Double.parseDouble(
                                String.valueOf(Optional.ofNullable(s.getAs("avgHsnsDay")).orElse(0D)));
                        int hesTimes = Integer.parseInt(
                                String.valueOf(Optional.ofNullable(s.getAs("hesTimes")).orElse(0)));
                        double ofeP = Double.parseDouble(
                                String.valueOf(Optional.ofNullable(s.getAs("ofeP")).orElse(0D)));
                        int brakeTimes = Integer.parseInt(
                                String.valueOf(Optional.ofNullable(s.getAs("brakeTimes")).orElse(0)));
                        double overspeedP = Double.parseDouble(
                                String.valueOf(Optional.ofNullable(s.getAs("overspeedP")).orElse(0D)));
                        entity.setTerminalId(terminalId);
                        entity.setmMilage(mMilage);
                        entity.setmGps(mMilage);
                        entity.setFuel(fuel);
                        entity.setWorkHours(workHours);
                        entity.setIdleHours(idleHours);
                        entity.setStboTimes(stboTimes);
                        entity.setAcceTimes(acceTimes);
                        entity.setOverspeedTimes(overspeedTimes);
                        entity.setHsnsTimes(hsnsTimes);
                        entity.setTdTimes(tdTimes);
                        entity.setFuelRateDay(fuelRateDay);
                        entity.setAvgSpeedNoidle(avgSpeedNoidle);
                        entity.setAvgSpeed(avgSpeed);
                        entity.setIdleP(idleP);
                        entity.setAvgHsnsDay(avgHsnsDay);
                        entity.setHesTimes(hesTimes);
                        entity.setOfeP(ofeP);
                        entity.setBrakeTimes(brakeTimes);
                        entity.setOverspeedP(overspeedP);
                        return entity;
                    })
                    .iterator();
        };
    }

    private static FlatMapFunction<Iterator<Tuple2<Long, Tuple2<CountEntity, CountEntity>>>, CountEntity> pariToCountEntityRDD() {
        return (FlatMapFunction<Iterator<Tuple2<Long, Tuple2<CountEntity, CountEntity>>>, CountEntity>) iterator ->
        {
            Iterable<Tuple2<Long, Tuple2<CountEntity, CountEntity>>> iterable = () -> iterator;
            return StreamSupport.stream(iterable.spliterator(), false)
                    .map(s -> {
                        CountEntity countEntity0200 = s._2._1;
                        CountEntity countEntityTrip = s._2._2;
                        countEntityTrip.setDate(countEntity0200.getDate());
                        countEntityTrip.setTimestamp(countEntity0200.getTimestamp());
                        countEntityTrip.setSpeedSum(countEntity0200.getSpeedSum());
                        countEntityTrip.setSpeedNum(countEntity0200.getSpeedNum());
                        countEntityTrip.setOilValue(countEntity0200.getOilValue());
                        countEntityTrip.setbLat(countEntity0200.getbLat());
                        countEntityTrip.setbLng(countEntity0200.getbLng());
                        countEntityTrip.seteLat(countEntity0200.geteLat());
                        countEntityTrip.seteLng(countEntity0200.geteLng());
                        countEntityTrip.setStart(countEntity0200.getStart());
                        countEntityTrip.setEnd(countEntity0200.getEnd());
                        countEntityTrip.setsDate(countEntity0200.getsDate());
                        return countEntityTrip;
                    })
                    .iterator();
        };
    }
}
